package com.ibm.cps.servlet.starter;

import java.io.IOException;
import java.util.List;

import org.apache.log4j.Logger;

import com.fasterxml.jackson.databind.node.ArrayNode;
import com.fasterxml.jackson.databind.node.ObjectNode;
import com.google.common.collect.Lists;
import com.ibm.cps.servlet.HttpClientMonitor;
import com.ibm.util.ErrorCode;
import com.ibm.util.JSONUtility;
import com.ibm.util.exception.CPSException;

public class YarnApplicationMonitor implements ProcessMonitor {
	private static final Logger logger = Logger.getLogger(YarnApplicationMonitor.class);
	private final String resourceManagerURL;
	
	public YarnApplicationMonitor(String resourceManagerURL) {
		this.resourceManagerURL = resourceManagerURL;
	}

	@Override
	public boolean isAlive(String tenantId) throws CPSException {
		String applicationId = getApplicationId(tenantId);
		return applicationId != null;
	}
	
	private String getApplicationId(String tenantId) throws CPSException{
		try{
			logger.info("connecting to YARN resource manager: " + resourceManagerURL);
			String request = HttpClientMonitor.sendGetRequest(resourceManagerURL);
			ObjectNode object = JSONUtility.fromObjectString(request);
			ArrayNode apps = (ArrayNode)object.get("apps").get("app");
			int appCount = apps.size();
			for(int i = 0; i < appCount; i++){
				ObjectNode appObject = (ObjectNode)apps.get(i);
				String name = appObject.get("name").asText();
				if(name.equals(tenantId)){
					String stateString = appObject.get("state").asText();
					YarnAppState state = YarnAppState.valueOf(stateString);
					if(runningStates.contains(state)){
						String applicationId = appObject.get("id").asText();
						return applicationId;
					}
				}
			}
			return null;
		}catch(IOException failedToConnect){
			failedToConnect.printStackTrace();
			throw new CPSException(ErrorCode.SERVER_NOT_FOUND_CODE, "Unable to connect to YARN resource manager URL " + resourceManagerURL);
		}catch (NullPointerException e) {
			// TODO: handle exception
			return null;
		}
	}
	
	private static enum YarnAppState {
		NEW, NEW_SAVING, SUBMITTED, ACCEPTED, RUNNING, FINISHED, FAILED, KILLED
	}
	
	private static List<YarnAppState> runningStates = Lists.newArrayList(YarnAppState.NEW, YarnAppState.NEW_SAVING, YarnAppState.SUBMITTED, YarnAppState.ACCEPTED, YarnAppState.RUNNING);

	@Override
	public void registerStarted(String tenantId, WorkerProcess process)
			throws CPSException {
	}

	@Override
	public void stop(String tenantId) throws CPSException {
		try{
			String applicationId = getApplicationId(tenantId);
			if(applicationId == null){
				logger.info("Spark on YARN application for tenant: " + tenantId + " is already stopped");
			}else{
				String command = "yarn application -kill " + applicationId;
				logger.info("spark on yarn stop command: " + command);
				Runtime.getRuntime().exec(command);
			}
		}catch (IOException ex){
			throw new CPSException(ErrorCode.INTERNAL_ERROR_CODE,
					"Failed to stop Spark on Yarn application");
		}
	}

}
